package org.luxor.common.netty;

import cn.hutool.core.thread.NamedThreadFactory;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.luxor.common.netty.config.ServerPipelineConfig;
import org.luxor.common.netty.config.properties.NettyServerProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Netty简单案例：非阻塞I/O案例 服务端
 *
 * @author Mr.yan @date 2023/8/29
 */
public class TcpNioServer extends Thread {
    private final Logger log = LoggerFactory.getLogger(this.getClass());

    /**
     * 当前网络通道
     */
    private ChannelFuture channelFuture;

    /**
     * 管理线程组
     */
    private EventLoopGroup bossGroup;

    /**
     * 工作线程组
     */
    private EventLoopGroup workerGroup;

    /**
     * 服务端配置
     */
    private final NettyServerProperties properties;

    /**
     * 处理器链配置
     */
    private final ChannelInitializer<Channel> channelInitializer;

    public TcpNioServer(ServerPipelineConfig config) {
        super("netty-server-boss-1");
        this.channelInitializer = config;
        this.properties = config.serverProperties();
    }

    @Override
    public void run() {
        try {
            log.info("[{}] bind port {}, Begin... ", properties.getId(), properties.getPort());
            doBind();
        } catch (Exception e) {
            log.error("[{}] bind port {}, Failed. ", properties.getId(), properties.getPort(), e);
        }
    }

    /**
     * 绑定端口
     */
    private void doBind() throws InterruptedException {
        // 连接管理线程组
        bossGroup = new NioEventLoopGroup(1, new NamedThreadFactory("netty-server-boss-", false));
        // 工作线程组
        workerGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2, new NamedThreadFactory("netty-server-worker-", true));
        try {
            // 配置服务端启动参数
            ServerBootstrap b = new ServerBootstrap().group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .option(ChannelOption.SO_REUSEADDR, true)
                    .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(32 * 1024, 64 * 1024))
                    .childHandler(this.channelInitializer);
            // 绑定端口，同步等待成功
            channelFuture = b.bind(this.properties.getPort()).sync();
            // 监听结果
            channelFuture.addListener(cf -> {
                boolean succeed = cf.isSuccess();
                if (succeed) {
                    log.info("[{}] bind port {}, Succeed.", properties.getId(), properties.getPort());
                } else {
                    log.error("[{}] bind port {}, Failed.", properties.getId(), properties.getPort());
                    channelFuture.channel().pipeline().fireChannelInactive();
                }
            });
            // 同步阻塞，等待服务端链路关闭
            channelFuture.channel().closeFuture().sync();
        } finally {
            // 优雅退出，释放线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    /**
     * 关闭服务
     */
    public void close() {
        try {
            if (channelFuture != null) {
                channelFuture.channel().close();
            }
            if (bossGroup != null) {
                bossGroup.shutdownGracefully();
            }
            if (workerGroup != null) {
                workerGroup.shutdownGracefully();
            }
        } catch (Throwable e) {
            log.error("[{}] Server stop error.", properties.getId(), e);
        }
    }
}
